package com.pajk.user.service.credit;

import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;

import com.alibaba.fastjson.JSON;
import com.pajk.user.dal.mapper.RedoMonitorMapper;
import com.pajk.user.model.RedoMonitor;
import com.pajk.user.service.common.idpool.IDPool;
import com.pajk.user.service.common.metaq.MsgObj;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.pajk.user.dal.mapper.CreditHisMapper;
import com.pajk.user.service.common.metaq.BaseConsumer;
import com.pajk.user.service.common.metaq.TopicEnum;

import javax.annotation.Resource;

/**
 * 
 * 万里通 同步积分
 * @author pshi
 *
 */
public class SynWLTCreditConsumer extends BaseConsumer {

	private static final Logger log = LoggerFactory.getLogger(SynWLTCreditConsumer.class);
	
	public static final String MSG_USERID = "userId";
	
	public static final String MSG_WANLIID = "wanliId";
	
	public static final String ACTION_ID = "actionId";
	
	public static final String msgSeparator = "_";
	
	@Resource
	private CreditHisMapper creditHisMapper;

    @Resource
    RedoMonitorMapper creditErrorMapper;
	
	@Resource
	private CreditServiceWlt creditServiceWlt;
	
	@Resource
	private TransactionTemplate transactionTemplate;

    @Resource
    private TransactionTemplate userTransactionTemplate;

    @Override
	public TopicEnum getTopicEnum() {
		return TopicEnum.USER_SYNC_CREDIT_TO_WLT;
	}

    @Override
    public void doLogErrorConsumeMessage(final MsgObj msgObj) {
        {
            List<Serializable> msgs=msgObj.getMsgList();
            for (final Serializable obj : msgs) {
                if (obj instanceof String) {
                    userTransactionTemplate.execute(new TransactionCallbackWithoutResult() {
                        public void doInTransactionWithoutResult(TransactionStatus status) {
                            try {
                                HashMap<String, Object> msg = (HashMap<String, Object>)JSON.parseObject((String)obj,HashMap.class);
                                String wanliId = (String) msg.get(MSG_WANLIID);

                                RedoMonitor creditError = new RedoMonitor();
                                creditError.setId(UUID.randomUUID().toString());
                                creditError.setBusinessId(wanliId);
                                creditError.setDescription("SynWLT Error");
                                creditError.setErrorMsg(msgObj.getErrorMsg());
                                creditError.setMsgBody(obj.toString());
                                creditError.setStatus(Byte.valueOf("1"));
                                creditError.setTopic(getTopicEnum().getTopic());
                                creditErrorMapper.insertRedoMonitor(creditError);
                            } catch (Exception e) {
                                log.error("doLogErrorConsumeMessage failed", e);
                                status.setRollbackOnly();
                            } catch (Throwable t){
                                log.error("throwable catched", t);
                                status.setRollbackOnly();
                            }
                        }
                    });
                }
            }
        }
    }

    @Override
    public ConsumeConcurrentlyStatus doConsumeMessage(final MsgObj msgObj) {
        final int maxRetryCount=super.maxRetryCount;
        List<Serializable> msgs=msgObj.getMsgList();
        for (final Serializable obj : msgs) {
            if (obj instanceof String) {
                return transactionTemplate.execute(new TransactionCallback<ConsumeConcurrentlyStatus>() {
                    public ConsumeConcurrentlyStatus doInTransaction(TransactionStatus status) { 
                            try {
                                HashMap<String, Object> msg = (HashMap<String, Object>) JSON.parseObject((String) obj, HashMap.class);

                                Long userId = (Long) msg.get(MSG_USERID);
                                String wanliId = (String) msg.get(MSG_WANLIID);
                                String actionId = (String) msg.get(ACTION_ID);
                                if (userId == null || wanliId == null) {
                                    log.error("synchroniz wlt failed, wanliId is null or userId is null");
                                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                                }

                                Map<String, Object> selectMap = new HashMap<String, Object>();
                                selectMap.put("sync", CreditStatusEnum.noWlt.value());
                                selectMap.put("userId", userId);
                                int credits = creditHisMapper.selectSumByUserIdAndSync(selectMap);

                                creditServiceWlt.syncWltCredit(wanliId, credits, actionId);

                                // sync 9 means synchronized
                                HashMap<String, Object> updateMap = new HashMap<String, Object>();
                                updateMap.put("sync", CreditStatusEnum.Synchronized.value());
                                updateMap.put("userId", userId);
                                updateMap.put("syncPre", CreditStatusEnum.noWlt.value());

                                creditHisMapper.updateCreditSyncByUserIdAndSync(updateMap);
                                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                            } catch(Throwable e){
                                log.error("sync wlt failed", e);
                                status.setRollbackOnly();
                                if(msgObj.getReconsumeTimes()>=maxRetryCount){
                                    log.error("reconsumeTimes >" + maxRetryCount + "msgs:" + msgObj + "context:" + msgObj.getContext());
                                    msgObj.setErrorMsg(e.getMessage());
                                    doLogErrorConsumeMessage(msgObj);
                                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                                }else{
                                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                                }

                        }
                    }
                });

            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

}
